package com.chm.fw.ui;

import android.os.Bundle;
import android.support.annotation.Nullable;
import android.util.Log;
import android.view.View;
import android.widget.TextView;

import com.chm.fw.R;
import com.chm.fw.base.BaseActivity;

import java.util.concurrent.TimeUnit;

import io.reactivex.Flowable;
import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.annotations.NonNull;
import io.reactivex.functions.Consumer;
import io.reactivex.schedulers.Schedulers;

public class RxActivity extends BaseActivity implements View.OnClickListener {

    private final String TAG = "RxActivity";

    private TextView tv;
    private String t;

    @Override
    protected void onCreate(@Nullable Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_rx);
        initView();
    }

    private void initView() {
        findViewById(R.id.a_rx_btn1).setOnClickListener(this);
        findViewById(R.id.a_rx_btn2).setOnClickListener(this);

        tv = findViewById(R.id.a_rx_tv);
    }

    @Override
    public void onClick(View view) {
        switch (view.getId()) {
            case R.id.a_rx_btn1:
                intervalFlowable();
                break;
            case R.id.a_rx_btn2:
                createObservable();
                break;
        }
    }

    private void intervalFlowable() {
        tv.setText("Flowable.interval(1000, TimeUnit.MILLISECONDS)\n");
        t = "";

        disposable = Flowable
                .interval(1000, TimeUnit.MILLISECONDS)
                .doOnNext(new Consumer<Long>() {
                    @Override
                    public void accept(@NonNull Long aLong) throws Exception {
                        Log.e(TAG, "doOnNext ：" + aLong);
                        t = tv.getText().toString() + "doOnNext:" + aLong + " (" + Thread.currentThread().getName() + ")\n";
                    }
                })
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<Long>() {
                    @Override
                    public void accept(@NonNull Long aLong) throws Exception {
                        Log.e(TAG, "subscribe ：" + aLong);
                        t = t + "subscribe:" + aLong + " (" + Thread.currentThread().getName() + ")\n";
                        tv.setText(t);
                        if (aLong >= 3) {
                            unSubscribe();
                        }
                    }
                });
    }

    private void createObservable() {
        tv.setText("Observable.create()\n");
        t = "";

        disposable = Observable
                /*发射线程*/
                .create(new ObservableOnSubscribe<Object>() {
                    @Override
                    public void subscribe(ObservableEmitter<Object> emitter) throws Exception {
                        Log.e(TAG, "create");
                        t = tv.getText().toString() + "create" + " (" + Thread.currentThread().getName() + ")\n";
                        emitter.onNext("666");
                        emitter.onComplete();
                    }
                })
                .subscribeOn(Schedulers.io())
                /*接收线程*/
                .observeOn(Schedulers.io())
                .doOnNext(new Consumer<Object>() {
                    @Override
                    public void accept(@NonNull Object o) throws Exception {
                        Log.e(TAG, "doOnNext ：" + o);
                        t = t + "doOnNext:" + o + " (" + Thread.currentThread().getName() + ")\n";
                    }
                })
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<Object>() {
                    @Override
                    public void accept(Object o) throws Exception {
                        Log.e(TAG, "subscribe ：" + o);
                        t = t + "subscribe:" + o + " (" + Thread.currentThread().getName() + ")\n";
                        tv.setText(t);
                    }
                });
    }
}
